#ifndef WIN32
#include <fnmatch.h>
#else
#include <Shlwapi.h>
#endif
#include <map>
#include <rise/threading/CriticalSection.h>
#include <rise/common/exmacros.h>
#include <rise/common/ExceptionTemplate.h>
#include "Message.h"
#include "MessageQueue.h"
#include "MessageQueueManager.h"

#ifdef WIN32
#define fnmatch(szPattern_fnmatch, szString_fnmatch, nFlags_fnMatch) \
  (PathMatchSpecA(szString_fnmatch, szPattern_fnmatch) == TRUE ? 0 : 1)
#endif

namespace smq
{
  struct MessageQueueManager::MessageQueueManagerImpl
  {
    struct MessageQueueCounter: public MessageQueue
    {
      int nCounter;
      MessageQueueCounter():
          nCounter(0)
      {
      }

      MessageQueueCounter& operator=(const MessageQueueCounter& rMessageQueue)
      {
        nCounter = rMessageQueue.nCounter;
        return *this;
      }

      MessageQueueCounter(const MessageQueueCounter& rMessageQueue)
      {
        nCounter = rMessageQueue.nCounter;
      }
    };

    struct QueueWatcherData
    {
      std::string sQueueName;
      QueueWatcher* pQueueWatcher;

      QueueWatcherData(const std::string& sQueueNameInit, QueueWatcher* pQueueWatcherInit):
        sQueueName(sQueueNameInit), pQueueWatcher(pQueueWatcherInit)
      {
      }

      bool operator==(const QueueWatcherData& rOther)
      {
        return sQueueName == rOther.sQueueName
            && pQueueWatcher == rOther.pQueueWatcher;
      }
    };

    MessageQueueManagerImpl():
      m_nDefaultQueueLimit(-1)
    {
    }

    typedef std::list<QueueWatcherData> QueueWatcherDataList;
    typedef std::map<std::string, MessageQueueCounter> MessageQueueMap;

    QueueWatcherDataList lsQueueWatchers;
    MessageQueueMap mMessageQueues;
    int m_nDefaultQueueLimit;
    rise::threading::CCriticalSection tLock;
  };

  MessageQueueManager::MessageQueueManager()
  {
    rise::LogEntry();
    m_pImpl = new MessageQueueManagerImpl;
  }

  MessageQueueManager::~MessageQueueManager()
  {
    rise::LogEntry();
    if (m_pImpl)
    {
      delete m_pImpl;
      m_pImpl = NULL;
    }
  }

  MessageQueueManager& MessageQueueManager::Inst()
  {
    rise::LogEntry();
    if (!m_pInst)
    {
      m_pInst = new MessageQueueManager;
    }

    return *m_pInst;
  }

  MessageQueue& MessageQueueManager::CreateQueue(const std::string& sQueueName)
  {
    rise::threading::CScopedCriticalSection(m_pImpl->tLock);
    RISE_ASSERTES(sQueueName != "", rise::CLogicParamException, "Queue name is empty");

    rise::LogEntry();
    rise::LogInfo() << "Creating queue: " << sQueueName;
    MessageQueueManagerImpl::MessageQueueCounter& rQueue = m_pImpl->mMessageQueues[sQueueName];
    if (rQueue.nCounter == 0)
    {
      rise::LogInfo() << "setting queue limit: " << m_pImpl->m_nDefaultQueueLimit;
      rQueue.SetQueueLimit(m_pImpl->m_nDefaultQueueLimit);
    }
    ++rQueue.nCounter;
    return rQueue;
  }

  void MessageQueueManager::RemoveQueue(const std::string& sQueueName)
  {
    rise::threading::CScopedCriticalSection(m_pImpl->tLock);
    RISE_ASSERTES(sQueueName != "", rise::CLogicParamException, "Queue name is empty");

    rise::LogInfo() << "Removing queue " << sQueueName;
    MessageQueueManagerImpl::MessageQueueMap::iterator itQueue = m_pImpl->mMessageQueues.find(sQueueName);
    RISE_ASSERTES(itQueue != m_pImpl->mMessageQueues.end(), rise::CLogicNoItemException, "Queue " + sQueueName + " does not exists");
    MessageQueueManagerImpl::MessageQueueCounter& rQueue = itQueue->second;

    if(!--rQueue.nCounter)
    {
      m_pImpl->mMessageQueues.erase(itQueue);
    }
  }

  MessageQueue& MessageQueueManager::GetQueue(const std::string& sQueueName)
  {
    rise::threading::CScopedCriticalSection(m_pImpl->tLock);
    RISE_ASSERTES(sQueueName != "", rise::CLogicParamException, "Queue name is empty");

    rise::LogInfo() << "Getting queue " << sQueueName;
    MessageQueueManagerImpl::MessageQueueMap::iterator itQueue = m_pImpl->mMessageQueues.find(sQueueName);
    RISE_ASSERTES(itQueue != m_pImpl->mMessageQueues.end(), rise::CLogicNoItemException, "Queue " + sQueueName + " does not exists");
    return itQueue->second;
  }

  void MessageQueueManager::GetQueueList(StringList& rlsQueues)
  {
    rlsQueues.clear();
    for (MessageQueueManagerImpl::MessageQueueMap::iterator itQueue = m_pImpl->mMessageQueues.begin();
      itQueue != m_pImpl->mMessageQueues.end(); ++itQueue)
    {
      rlsQueues.push_back(itQueue->first);
    }
  }

  void MessageQueueManager::GetFilteredQueueList(const std::string& sFilter, StringList& rlsQueues)
  {
    rlsQueues.clear();
    for (MessageQueueManagerImpl::MessageQueueMap::iterator itQueue = m_pImpl->mMessageQueues.begin();
      itQueue != m_pImpl->mMessageQueues.end(); ++itQueue)
    {
      if (fnmatch(sFilter.c_str(), itQueue->first.c_str(), 0) == 0)
      {
        rlsQueues.push_back(itQueue->first);
      }
    }
  }

  void MessageQueueManager::AddQueueWatcher(const std::string& sQueueName, QueueWatcher* pQueueWatcher)
  {
    rise::threading::CScopedCriticalSection(m_pImpl->tLock);
    RISE_ASSERTES(sQueueName != "", rise::CLogicParamException, "Queue name is empty");
    rise::LogDebug1() << "Adding queue watcher " << pQueueWatcher << " to queue " << sQueueName;
    m_pImpl->lsQueueWatchers.push_back(MessageQueueManagerImpl::QueueWatcherData(sQueueName, pQueueWatcher));
  }

  void MessageQueueManager::RemoveQueueWatcher(const std::string& sQueueName, QueueWatcher* pQueueWatcher)
  {
    rise::threading::CScopedCriticalSection(m_pImpl->tLock);
    RISE_ASSERTES(sQueueName != "", rise::CLogicParamException, "Queue name is empty");
    rise::LogDebug1() << "Removing queue watcher " << pQueueWatcher << " to queue " << sQueueName;
    m_pImpl->lsQueueWatchers.remove(MessageQueueManagerImpl::QueueWatcherData(sQueueName, pQueueWatcher));
  }

  void MessageQueueManager::RemoveQueueWatcher(QueueWatcher* pQueueWatcher)
  {
    rise::threading::CScopedCriticalSection(m_pImpl->tLock);
    rise::LogDebug1() << "Removing watcher " << pQueueWatcher << " from all queues";

    for (MessageQueueManagerImpl::QueueWatcherDataList::iterator itWatcher = m_pImpl->lsQueueWatchers.begin();
          itWatcher != m_pImpl->lsQueueWatchers.end(); )
    {
      if (itWatcher->pQueueWatcher == pQueueWatcher)
      {
        rise::LogDebug1() << "Removing watcher " << pQueueWatcher << " from queue(s) " << itWatcher->sQueueName;
        m_pImpl->lsQueueWatchers.erase(itWatcher++);
      }
      else
      {
        ++itWatcher;
      }
    }
  }

  void MessageQueueManager::AddMessage(Message& rMsg)
  {
    rise::threading::CScopedCriticalSection tLock(GetLock());

    // publishing message to queues
    bool bFound = false;
    RISE_ASSERTES(rMsg.sQueue != "", rise::CLogicParamException, "Queue name is empty");

    for (MessageQueueManagerImpl::MessageQueueMap::iterator itQueue = m_pImpl->mMessageQueues.begin();
         itQueue != m_pImpl->mMessageQueues.end(); ++itQueue)
    {
      if (fnmatch(rMsg.sQueue.c_str(), itQueue->first.c_str(), 0) == 0)
      {
        rise::LogDebug1() << "Adding message to queue " << itQueue->first << " by name " << rMsg.sQueue;
        itQueue->second.AddMessage(rMsg);
        bFound = true;
      }
    }
    RISE_ASSERTES(bFound, rise::CLogicNoItemException, "Queue(s) " + rMsg.sQueue + " does not exists");

    // sending message to subscribers
    for (MessageQueueManagerImpl::QueueWatcherDataList::iterator itWatcher = m_pImpl->lsQueueWatchers.begin();
          itWatcher != m_pImpl->lsQueueWatchers.end(); ++itWatcher)
    {
      if (fnmatch(itWatcher->sQueueName.c_str(), rMsg.sQueue.c_str(), 0) == 0)
      {
        rise::LogDebug1() << "Forwarding message to watcher's " << itWatcher->pQueueWatcher
            << " queue " << itWatcher->sQueueName << " by name " << rMsg.sQueue;
        (*(itWatcher->pQueueWatcher))(rMsg);
      } else
      if (fnmatch(rMsg.sQueue.c_str(), itWatcher->sQueueName.c_str(), 0) == 0)
      {
        rise::LogDebug1() << "Forwarding message to watcher's " << itWatcher->pQueueWatcher
            << "queue " << itWatcher->sQueueName << " by name " << rMsg.sQueue;
        // changing queue name because it should'nt be a regex
        Message tMsg = rMsg;
        tMsg.sQueue = itWatcher->sQueueName;
        (*(itWatcher->pQueueWatcher))(tMsg);
      }
    }
  }

  void MessageQueueManager::SetDefaultQueueLimit(int nLimit /*= 100*/)
  {
    m_pImpl->m_nDefaultQueueLimit = nLimit;
  }

  rise::threading::CCriticalSection& MessageQueueManager::GetLock()
  {
    return m_pImpl->tLock;
  }


  MessageQueueManager* MessageQueueManager::m_pInst = NULL;
}
